package com.spark.lola.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.spark.lola.logistics.constant.LogisticsConstant;
import com.spark.lola.logistics.entity.Message;
import com.spark.lola.logistics.entity.OrderInfo;
import com.spark.lola.logistics.entity.OrderProcess;
import com.spark.lola.logistics.service.MessageService;
import com.spark.lola.logistics.vo.OrderStatisticVo;
import com.spark.lola.support.common.entity.ShiroUser;
import com.spark.lola.support.wms.cache.CollectionCache;
import com.spark.lola.support.wms.configuration.JPushProperties;
import com.spark.lola.support.wms.configuration.WmsProperties;
import com.spark.lola.support.wms.constant.KafkaTopic;
import com.spark.lola.support.wms.constant.WmsNettyCommands;
import com.spark.lola.support.wms.dto.JPushSendDto;
import com.spark.lola.support.wms.dto.OrderDto;
import com.spark.lola.support.wms.util.JPushUtil;
import com.spark.lola.support.wms.vo.NettyOrderMessage;
import com.spark.lola.support.netty.common.NettyCacheUtils;
import com.spark.lola.support.netty.service.HawkPushServiceApi;
import io.netty.channel.Channel;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @author: Zhi-peng wang
 * @date: 2020/05/13 10:14
 * 货运单消费
 **/
@Component
public class LogisticsOrderConsumer {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private WmsProperties wmsProperties;

    @Autowired
    private HawkPushServiceApi pushServiceApi;

    @Autowired
    private MessageService messageService;

    @Autowired
    private JPushProperties jPushProperties;

    @Autowired
    private JPushUtil jPushUtil;

    @KafkaListener(topics = KafkaTopic.LOGISTICS_ORDER_CHANGE)
    @Transactional(rollbackFor = Exception.class)
    public void change(ConsumerRecord<String, String> consumerRecord) throws Exception {
        logger.info("===consumer【LogisticsOrder】货运单【{}】===START===", consumerRecord.key());
        OrderDto<JSONObject> model = JSON.parseObject(consumerRecord.value(), OrderDto.class);
        OrderProcess orderProcess = JSON.parseObject(model.getMessage(), OrderProcess.class);
        OrderInfo orderInfo = JSON.parseObject(model.getOrder().toJSONString(), OrderInfo.class);

        // 拼装通知消息
        NettyOrderMessage message = new NettyOrderMessage();
        message.setOrderId(orderInfo.getId().toString());
        message.setOrderNo(orderInfo.getNo());
        message.setStatus(orderProcess.getActionCode());
        message.setMessage(orderProcess.getInfo());

        // 极光推送给驾驶员
        JPushSendDto sendDto = new JPushSendDto();
        sendDto.setTitle(orderProcess.getActionName());
        sendDto.setContent(model.getJPushMessage());
        sendDto.setPushConditions(Collections.singletonList(orderInfo.getCreatorId().toString()));
        jPushUtil.sendPushWithCallback(jPushProperties.getLogistics(), sendDto);
        // 写入消息
        messageService.save(new Message(orderInfo.getCreatorId(), "运输单号：" + orderInfo.getNo(), model.getJPushMessage()));

        logger.info("===推送消息");
        dealPush(orderInfo, orderProcess, message);
        logger.info("===consumer【LogisticsOrder】货运单【{}】===END===", consumerRecord.key());
    }

    private void dealPush(OrderInfo orderInfo, OrderProcess orderProcess, NettyOrderMessage message) {
        Set<Channel> channelSet = new HashSet<>();
        // 推送给超级管理员
        List<ShiroUser> supers = CollectionCache.dutyEmployees.get(1L);
        if (!CollectionUtils.isEmpty(supers)) {
            supers.forEach(superUser -> {
                Set<Channel> channels = NettyCacheUtils.getChannel(superUser.getId().toString());
                if (!CollectionUtils.isEmpty(channels)) {
                    channelSet.addAll(channels);
                }
            });
        }
        // 推送给驾驶员
        Set<Channel> driverChannelSet = NettyCacheUtils.getChannel(String.valueOf(orderInfo.getCreatorId()));
        if (!CollectionUtils.isEmpty(driverChannelSet)) {
            channelSet.addAll(driverChannelSet);
        }
        // 推送给物流主管
        List<ShiroUser> logistics = CollectionCache.dutyEmployees.get(wmsProperties.getLogistics().getDutyId().get(0));
        if (!CollectionUtils.isEmpty(logistics)) {
            logistics.forEach(lead -> {
                Set<Channel> leadChannelSet = NettyCacheUtils.getChannel(lead.getId().toString());
                if (!CollectionUtils.isEmpty(leadChannelSet)) {
                    channelSet.addAll(leadChannelSet);
                }
            });
        }
        // 定价发送至财务主管
        if (message.getStatus().equals(LogisticsConstant.OrderStatus.ALREADY_GATHERING)) {
            List<ShiroUser> finance = CollectionCache.dutyEmployees.get(wmsProperties.getLogistics().getDutyId().get(2));
            finance.forEach(lead -> {
                Set<Channel> leadChannelSet = NettyCacheUtils.getChannel(lead.getId().toString());
                if (!CollectionUtils.isEmpty(leadChannelSet)) {
                    channelSet.addAll(leadChannelSet);
                }
            });
        }
        if (!CollectionUtils.isEmpty(channelSet)) {
            pushServiceApi.pushMsg(channelSet, WmsNettyCommands.PUSH_ORDER_CHANGE_LOGISTICS, message.toString());
            if (orderProcess.getActionCode() == LogisticsConstant.OrderStatus.ALREADY_SHIPMENTS &&
                    orderInfo.getIsRejected() == LogisticsConstant.IsRejected.NO) {
                OrderStatisticVo orderStatisticVo = new OrderStatisticVo();
                orderStatisticVo.setCreatorId(orderInfo.getCreatorId());
                orderStatisticVo.setLineType(orderInfo.getLineType());
                orderStatisticVo.setCarNo(orderInfo.getCarNo());
                orderStatisticVo.setStartArea(orderInfo.getStartArea());
                orderStatisticVo.setStartAddress(orderInfo.getStartAddress());
                orderStatisticVo.setShipmentsTime(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(orderProcess.getCreateTime()));
                pushServiceApi.pushMsg(channelSet, WmsNettyCommands.PUSH_LOGISTICS_ORDER, JSON.toJSONString(orderStatisticVo));
            }
        }
    }

}
